#pragma once

#include "index/index.h"
#include "index/inverted_index/inverted_indexer.h"
#include "file_system/file_reader/block_file_reader.h"
#include "file_system/file_reader/mmap_file_reader.h"

namespace hawking {
namespace indexlib {

struct InvertedTermUpdator {
public:
    InvertedTermUpdator() : valid_(false) {}

    void Init(uint64_t t, size_t bitmap_size) {
        term_key_ = t;
        if (!atomic_bitmap_) {
            atomic_bitmap_ = std::make_unique<AtomicBitmap>(bitmap_size);
        }
    }

    uint64_t TermKey() const {
        return term_key_;
    }

    void Set(DocID doc_id) {
        atomic_bitmap_.Set(doc_id);
    }

    void Clear(DocID doc_id) {
        atomic_bitmap_.Clear(doc_id);
    }

    bool Test(DocID doc_id) const {
        return atomic_bitmap_.Test(doc_id);
    }

    void Traverse(std::vector<DocId>* result) const {
        atomic_bitmap_.Traverse(result);
    }

    void TraverseTopk(std::vector<DocId>* result, uint32_t topk) const {
        atomic_bitmap_.TraverseTopk(result, topk);
    }

    bool IsValid() const {
        return valid_.load(std::memory_order_acq_rel);
    }

    void SetValid() const {
        valid_.store(true, std::memory_order_release);
    }

private:
    std::atomic<bool> valid_;
    uint64_t term_key_ = 0;
    std::unique_ptr<AtomicBitmap> atomic_bitmap_;
};

class InvertedDiskIndexer : public InvertedIndexer {
public:
    static const uint32_t term_key_length_ = sizeof(uint64_t);
    static const uint32_t offset_in_datafile_length_ = sizeof(uint64_t);
    static const uint32_t doclist_length_ = sizeof(uint32_t);
    static const uint32_t offset_file_item_bytes_ =
        term_key_length_ + offset_in_datafile_length_ + doclist_length_;

    class TermDiskIterator : public TermIterator {
    public:
        TermDiskIterator(const MmapFileReader& inverted_offset_reader) :
            inverted_offset_reader_(&inverted_offset_reader) {}
        virtual ~TermDiskIterator() = default;

        virtual bool HasNext() const override {
            return inverted_offset_reader_ &&
                current_offset_ < inverted_offset_reader_->Length();
        }

        virtual void Next() {
            current_offset_ += offset_file_item_bytes_;
        }

        virtual void Value(uint64_t* term) const {
            pk_disk_reader_->Read((uint8_t*)term, sizeof(*term), current_offset_);
        }

    private:
        const MmapFileReader* inverted_offset_reader_ = nullptr;
        size_t current_offset_ = 0;
    };

    InvertedDiskIndexer() = default;
    virtual ~InvertedDiskIndexer() = default;

    Status Load(
        const std::string& index_dir, bool need_increase, FSFileType data_filetype);

    virtual Status Lookup(
        uint64_t term_key, uint32_t topk, SegmentContext* segment_ctx) override;

    uint64_t TermCount() const {
        return term_count_;
    }

    virtual Status UpdateDocument(
        uint64_t old_term, uint64_t new_term, DocId doc_id) override;

    std::shared_ptr<TermDiskIterator> CreateIterator() const {
        return std::make_shared<TermDiskIterator>(*offset_file_reader_);
    }

    uint64_t DocCount() const {
        return doc_count_;
    }

private:
    Status AddDocument_(uint64_t new_term, doc_id);

    Status Lookup_(
        uint64_t offset_in_datafile,
        uint32_t topk,
        SegmentContext* segment_ctx,
        bool need_doc_context = true);

    std::unique_ptr<FileReader> data_file_reader_;
    std::unique_ptr<MmapFileReader> offset_file_reader_;

    bool need_increase_ = false;
    std::vector<InvertedTermUpdator> terms_updator_;
    uint64_t fix_term_count_ = 0;
    uint64_t dynamic_term_count_ = 0;

    uint64_t doc_count_ = 0;
    uint64_t min_term_key_ = 0;
    uint64_t max_term_key_ = 0;
};

}
}